package com.chiaic.unittest.unit;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

@Data
public class KafkaQueue<E>  {

    Properties propsProducer = new Properties();

    Properties propsConsumer = new Properties();

    Producer<String,String> producer;
    KafkaConsumer<String, String> consumer;
    String bootstrap;
    int lingerMs;



    public KafkaQueue(String bootstrap, int lingerMs) {
        this.bootstrap = bootstrap;
        this.lingerMs = lingerMs;

        propsProducer.put("bootstrap.servers", bootstrap);
        propsProducer.put("linger.ms", lingerMs);
        propsProducer.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        propsProducer.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer<String, String>(propsProducer);
        propsConsumer.put("bootstrap.servers", bootstrap);
        propsConsumer.setProperty("group.id", "test");
        propsConsumer.setProperty("enable.auto.commit", "true");
        propsConsumer.setProperty("auto.commit.interval.ms", "1000");
        propsConsumer.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        propsConsumer.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<String, String>(propsConsumer);
        consumer.subscribe(Arrays.asList("messagePackage"));
    }

    public boolean add(E o) {
        producer.send(new ProducerRecord<String, String>("messagePackage",o.toString() ));
        return true;
    }

    public E poll() {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        return null;
    }

    public E poll(long timeout, TimeUnit unit)  {
        return null;
    }

}
